package com.zyh.flink.day07.trigger;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class ContinuousProcessingTimeTriggerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = environment.socketTextStream("hadoop10", 9998);

        AllWindowedStream<String, TimeWindow> windowedStream = dataStream.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                /*
                * 每个窗口大小为20s
                * 每隔3s触发一次
                * 触发6次,最后2s的数据回丢失
                * */
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(3)));

        SingleOutputStreamOperator<String> result = windowedStream.process(new ProcessAllWindowFunction<String, String, TimeWindow>() {
            @Override
            public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {

                TimeWindow window = context.window();
                long start = window.getStart();
                long end = window.getEnd();
                long currentTime = System.currentTimeMillis();
                out.collect("start=" + start + ",end=" + end + ",currentTime=" + currentTime+",elements="+elements);

            }
        });

        result.print();

        environment.execute("continuousProcessingTimeTrigger");
    }
}
